package site.teamo.biu.flink.web.service.impl;

import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.IoUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.crypto.digest.DigestUtil;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
import site.teamo.biu.flink.common.function.descriptor.OperatorDescriptor;
import site.teamo.biu.flink.common.function.descriptor.OperatorPackageDescriptor;
import site.teamo.biu.flink.web.command.CommandHandler;
import site.teamo.biu.flink.web.configuration.BiuFlinkConfig;
import site.teamo.biu.flink.web.entity.OperatorDO;
import site.teamo.biu.flink.web.entity.OperatorPackageDO;
import site.teamo.biu.flink.web.exception.BiuResponseCode;
import site.teamo.biu.flink.web.mapper.FunctionMapper;
import site.teamo.biu.flink.web.mapper.FunctionPackageMapper;
import site.teamo.biu.flink.web.service.OperatorPackageService;
import site.teamo.biu.flink.web.service.model.functionpackage.VerifyFunctionPackageResult;

import javax.annotation.Resource;
import java.io.File;
import java.io.IOException;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.jar.JarEntry;
import java.util.jar.JarFile;
import java.util.stream.Collectors;

/**
 * @author haocongshun
 * @date 2023/10/30 17:27:15
 */
@Service
public class OperatorPackageServiceImpl extends ServiceImpl<FunctionPackageMapper, OperatorPackageDO> implements OperatorPackageService {

    @Resource
    private BiuFlinkConfig biuFlinkConfig;

    @Resource
    private CommandHandler commandHandler;

    @Resource
    private FunctionMapper functionMapper;

    @Override
    public List<OperatorPackageDO> queryList(String keyword) {
        LambdaQueryWrapper<OperatorPackageDO> queryWrapper = new LambdaQueryWrapper<OperatorPackageDO>()
                .select(
                        OperatorPackageDO::getId,
                        OperatorPackageDO::getName,
                        OperatorPackageDO::getTag,
                        OperatorPackageDO::getOperatorNumber,
                        OperatorPackageDO::getMd5,
                        OperatorPackageDO::getCreateTime
                );
        if (StrUtil.isNotBlank(keyword)) {
            keyword = "%" + keyword + "%";
            queryWrapper.like(OperatorPackageDO::getName, keyword)
                    .or()
                    .like(OperatorPackageDO::getPackageInfo, keyword)
                    .or()
                    .like(OperatorPackageDO::getTag, keyword);
        }
        return list(queryWrapper);
    }

    @Override
    public List<OperatorDescriptor> info(String id) {
        LambdaQueryWrapper<OperatorPackageDO> queryWrapper = new LambdaQueryWrapper<OperatorPackageDO>()
                .select(OperatorPackageDO::getPackageInfo)
                .eq(OperatorPackageDO::getId, id);
        OperatorPackageDO functionPackageDO = getOne(queryWrapper);
        if (functionPackageDO == null) {
            throw BiuResponseCode.GENERAL_ERROR.runtimeException("未查询到算子包信息");
        }
        OperatorPackageDescriptor descriptor = OperatorPackageDescriptor.generate(functionPackageDO.getPackageInfo());
        return descriptor.getOperators();
    }

    @Override
    public VerifyFunctionPackageResult verify(MultipartFile file) {
        String functionPackagePath = biuFlinkConfig.getFunctionPackagePath();
        try {
            String tempPath = functionPackagePath + File.separator + "temp-" + System.currentTimeMillis() + "-" + file.getOriginalFilename();
            file.transferTo(new File(tempPath));
            try (JarFile jar = new JarFile(tempPath)) {
                JarEntry functionInfoEntry = jar.getJarEntry("META-INF/function-info.json");
                if (functionInfoEntry == null) {
                    throw BiuResponseCode.GENERAL_ERROR.runtimeException("未发现算子包描述文件:META-INF/function-info.json");
                }
                //获取算子包描述信息
                String functionInfoStr = IoUtil.readUtf8(jar.getInputStream(functionInfoEntry));
                //根据描述信息验证算子包内容
                OperatorPackageDescriptor descriptor = OperatorPackageDescriptor.generate(functionInfoStr);
                if (CollectionUtil.isEmpty(descriptor.getOperators())) {
                    throw BiuResponseCode.GENERAL_ERROR.runtimeException("描述文件未记录任何算子");
                }
                Map<String, String> functionClassMap = descriptor.getOperators().stream()
                        .collect(Collectors.toMap(OperatorDescriptor::getClassName, OperatorDescriptor::getType));
                Map<String, String> verifyFunctionPackageResult = commandHandler.verifyFunctionPackage(functionClassMap, tempPath);
                return new VerifyFunctionPackageResult(tempPath, descriptor, verifyFunctionPackageResult);
            }
        } catch (IOException e) {
            throw BiuResponseCode.GENERAL_ERROR.runtimeException("算法包上传失败", e);
        }
    }

    @Transactional(rollbackFor = Exception.class)
    @Override
    public OperatorPackageDO save(String tempFilePath, OperatorPackageDescriptor descriptor) {
        if (!FileUtil.exist(tempFilePath)) {
            throw BiuResponseCode.GENERAL_ERROR.runtimeException("未发现算子包临时文件");
        }
        String tag = DateUtil.format(new Date(), "yy.MM.dd.HHmmssSSS");
        String filePath = biuFlinkConfig.getFunctionPackagePath() + File.separator + descriptor.getName() + "-" + tag + ".jar";
        FileUtil.rename(new File(tempFilePath), filePath, true);
        OperatorPackageDO functionPackageDO = new OperatorPackageDO()
                .setId(generateId(descriptor.getName(), tag))
                .setName(descriptor.getName())
                .setTag(tag)
                .setPath(filePath)
                .setMd5(DigestUtil.md5Hex(new File(filePath)))
                .setOperatorNumber(descriptor.getOperators().size())
                .setPackageInfo(JSON.toJSONString(descriptor));
        for (OperatorDescriptor function : descriptor.getOperators()) {
            String id = DigestUtil.md5Hex(functionPackageDO.getId() + function.getClassName());
            OperatorDO functionDO = new OperatorDO()
                    .setId(id)
                    .setOperatorPackageId(functionPackageDO.getId())
                    .setName(function.getName())
                    .setClassName(function.getClassName())
                    .setType(function.getType())
                    .setParameters(JSON.toJSONString(function.getOperatorParameters()));
            functionMapper.insert(functionDO);
        }
        this.save(functionPackageDO);
        return functionPackageDO;
    }

    @Override
    public void delete(List<String> ids) {
        List<OperatorPackageDO> list = list(new LambdaQueryWrapper<OperatorPackageDO>()
                .select(OperatorPackageDO::getId, OperatorPackageDO::getPath)
                .in(OperatorPackageDO::getId, ids));
        getBaseMapper().deleteBatchIds(ids);
        for (OperatorPackageDO functionPackageDO : list) {
            FileUtil.del(functionPackageDO.getPath());
        }
    }

    private String generateId(String name, String tag) {
        return DigestUtil.md5Hex(name + tag);
    }
}
